package main

import (
	"bufio"
	"fmt"
	"io"
	"os"
	"strings"
	"time"

	log "github.com/kpango/glg"
	crono "github.com/utahta/go-cronowriter"
)

type Controller struct {
	Kafka        KafkaHandler
	Config       *Configure
	MetricMapper Mapper
	Logger       string
	closing      chan interface{}
	closed       chan interface{}
}

func NewController(logger, conf, mapp string) (*Controller, error) {
	c := Controller{Logger: logger}
	c.BuildLogger()
	mapper, err := LoadMetricMapper(mapp)
	if err != nil {
		log.Printf("LoadMetricMapper: %v", err)
		return &c, err
	}
	c.MetricMapper = mapper
	config, err := LoadConfig(conf)
	if err != nil {
		log.Printf("LoadConfig: fail to load configure: %v", err)
		return &c, err
	}
	c.Config = config
	kh, err := NewKafkaHandler(*KafkaUrls, *Topic)
	if err != nil {
		log.Printf("NewKafkaHandler: fail to create kafka producer: %v", err)
		return &c, err
	}
	c.Kafka = kh
	return &c, nil
}

func (c *Controller) BuildLogger() {
	logger := crono.MustNew(c.Logger+".%Y%m", crono.WithInit())
	log.Get().SetMode(log.WRITER).AddWriter(logger)
}

func (c *Controller) Serve() {
	tick := time.Tick(c.Config.Interval)
	for {
		select {
		case <-tick:
			err := c.ScrapeAndUpload(c.Config.PromUrls)
			if err != nil {
				log.Errorf("%v", err)
			}
		case <-c.closing:
			err := c.Kafka.Close()
			if err != nil {
				log.Errorf("kafka close: %v", err)
			}
			return
		}
	}
}

type Mapper map[string]map[string]string

func LoadMetricMapper(file string) (Mapper, error) {
	mapper := make(Mapper)
	lineNum := 0
	err := ReadAndHandleLine(file, func(line string) error {
		lineNum++
		if strings.Contains(line, "#") {
			return nil
		}
		m := strings.Split(line, ",")
		if len(m) != 3 {
			return fmt.Errorf("parsing error in line %d", lineNum)
		}
		metric, subMetric, goliaMetric := m[0], m[1], m[2]
		if mapper[metric] == nil {
			mapper[metric] = make(map[string]string)
		}
		mapper[metric][subMetric] = goliaMetric
		return nil
	})
	return mapper, err
}

func ReadAndHandleLine(fileName string, handler func(string) error) error {
	f, err := os.Open(fileName)
	if err != nil {
		return err
	}
	defer f.Close()
	buf := bufio.NewReader(f)
	for {
		line, err := buf.ReadString('\n')
		if err == io.EOF {
			break
		}
		if err != nil {
			return err
		}
		line = strings.TrimSpace(line)
		err = handler(line)
		if err != nil {
			return err
		}
	}
	return nil
}

func (c *Controller) ScrapeAndUpload(promUrls []string) error {
	for _, url := range promUrls {
		data, err := Scrape(url)
		if err != nil {
			return fmt.Errorf("scrape: %v", err)
		}
		for _, metric := range data {
			record, err := Transform(metric, c.MetricMapper)
			if err != nil {
				log.Printf("transform: %v", err)
				continue
			}
			offset, err := c.Kafka.Produce(record)
			if err != nil {
				return fmt.Errorf("kafka produce: %v", err)
			} else {
				logmsg := "kafka produce: produced message successfully. Metric:%s, Timestamp:%v, Offset:%v"
				log.Printf(logmsg, metric.Metric, metric.Timestamp, offset)
			}
		}
	}
	return nil
}

func (c *Controller) Close() {
	close(c.closing)
	<-c.closed
}
